package cn.hlyc.mq.transaction;


import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.rmi.MarshalException;
import java.util.HashMap;

/**
 * 生产者
 */

/**
 *
 * 1.发送事务消息 创建的生产者是TransactionMQProducer
 * 2.需要设置事务的监听器setTransactionListener
 *      监听器的写法:
 *              实现TransactionListener 重写两个方法
 *              executeLocalTransaction: 执行本地事务
 *                      两个事务 一个是 message(发给消费者的消息)
 *                              一个是 一个object(就是发送事务消息的时候 传来的object)
 *                      返回值: COMMIT_MESSAGE(成功) ROLLBACK_MESSAGE(失败回滚)
 *              checkLocalTransaction：检查本地事务是否执行成功
 *                      返回值 :COMMIT_MESSAGE(成功) ROLLBACK_MESSAGE(失败回滚)
 * 3.发送消息的时候 调用的方法是:sendMessageInTransaction
 *          传入的两个参数
 *                 message 消息
 *                 obj 就是执行本地事务的方法的第二个参数
 *
 */

public class TransactionProducer {
    public static void main(String[] args)throws Exception {
        //创建生产者实例
        TransactionMQProducer producer = new TransactionMQProducer("mq-tran-producer-group");
        //设置mq的nameserver地址
        producer.setNamesrvAddr("127.0.0.1:9876");

        producer.setTransactionListener(new MyTransactionListener());
        //启动生产者
        producer.start();

        byte[] msgByteArray = "事务消息".getBytes(RemotingHelper.DEFAULT_CHARSET);

        Message message = new Message("tran-topic", "tran-tag", msgByteArray);

        HashMap<String, String> objectObjectHashMap = new HashMap<>();
        objectObjectHashMap.put("1","4");
        objectObjectHashMap.put("2","5");
        objectObjectHashMap.put("3","6");

        //发消息
        TransactionSendResult sendResult = producer.sendMessageInTransaction(message,objectObjectHashMap);
        System.out.println(sendResult);

        //关闭资源
        producer.shutdown();
    }
}
















